package com.atguigu.flink.chapter11;

import com.atguigu.flink.chapter05.Source.WaterSensor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.AggregateFunction;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/3/12 23:54
 */
public class Flink18_UDAF {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        SingleOutputStreamOperator<WaterSensor> sensorDS = env
                .socketTextStream("localhost", 9999)
                .map(line -> {
                    String[] split = line.split(",");
                    return new WaterSensor(split[0],
                            Long.parseLong(split[1]),
                            Integer.parseInt(split[2]));
                });


        Table table = tableEnv.fromDataStream(sensorDS);


        // 先注册
        tableEnv.createTemporaryFunction("myAvg", MyAvg.class);

        // TableAPI
//        table
//                .groupBy($("id"))
//                .select($("id"),call("myAvg",$("vc")))
//                .execute()
//                .print();



        // SQL
        tableEnv.executeSql("select id,myAvg(vc) from "+table + " group by id").print();


    }

    public static class MyAvg extends AggregateFunction<Double, SumCount> {

        @Override
        public Double getValue(SumCount accumulator) {
            return accumulator.getVcSum() * 1D / accumulator.getVcCount();
        }

        public void accumulate(SumCount acc, Integer vc) {
            acc.setVcSum(acc.getVcSum() + vc);
            acc.setVcCount(acc.getVcCount() + 1);
        }

        @Override
        public SumCount createAccumulator() {
            return new SumCount();
        }
    }
}
